package com.hmdp.service.impl;

import cn.hutool.core.bean.BeanUtil;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.hmdp.dto.Result;
import com.hmdp.entity.SeckillVoucher;
import com.hmdp.entity.VoucherOrder;
import com.hmdp.mapper.VoucherMapper;
import com.hmdp.mapper.VoucherOrderMapper;
import com.hmdp.service.IVoucherOrderService;
import com.hmdp.utils.RedisConstants;
import com.hmdp.utils.RedisIdWork;
import com.hmdp.utils.UserHolder;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.aop.framework.AopContext;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.time.Duration;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.locks.ReentrantLock;

/**
 * <p>
 * 服务实现类
 * </p>
 *
 * @author 虎哥
 * @since 2021-12-22
 */
@Slf4j
@Service
public class VoucherOrderServiceImpl extends ServiceImpl<VoucherOrderMapper, VoucherOrder> implements IVoucherOrderService {

    @Resource
    private VoucherOrderMapper voucherOrderMapper;
    @Resource
    private SeckillVoucherServiceImpl seckillVoucherService;
    @Resource
    private StringRedisTemplate stringRedisTemplate;
    @Resource
    private VoucherMapper voucherMapper;
    @Resource
    private RedisIdWork redisIdWork;
    @Resource
    private RedissonClient redissonClient;

    private static DefaultRedisScript<Long> CREATE_ORDER;

    static {
        CREATE_ORDER = new DefaultRedisScript<>();
        CREATE_ORDER.setLocation(new ClassPathResource("lua/createSkillOrder.lua"));
        CREATE_ORDER.setResultType(Long.class);
    }

    private ReentrantLock skillLock = new ReentrantLock();
    // 阻塞队列
    private BlockingQueue<VoucherOrder> orderQueue = new ArrayBlockingQueue<>(1024 * 1024);
    // 线程池
    private static final ExecutorService FINISH_CREATE_ORDER = Executors.newSingleThreadExecutor();
    // 代理对象
    private IVoucherOrderService proxy;

    /**
     * @postConstruct 注解
     * 类初始化完成后执行该方法
     */
    @PostConstruct
    public void init() {
        FINISH_CREATE_ORDER.submit(new CreateOrder());
    }

    private class CreateOrder implements Runnable {
        /**
         * 阻塞队列 异步处理
         * @param voucherOrder
         */
//        @Override
//        public void run() {
//            while (true) {
//                try {
//                    VoucherOrder voucherOrder = orderQueue.take();
//                    // 执行创建订单业务
//                    handleCreateOrder(voucherOrder);
//                } catch (InterruptedException e) {
//                    log.error("处理订单异常{}", e.getMessage());
//                }
//            }
//        }

        /**
         * redis 消息队列异步处理
         */
        @Override
        public void run() {
            while (true) {
                try {
                    handlerMessageList();
                } catch (Exception e) {
                    // 处理消息出现异常时 从(PendingList)消费者已取但未确认的消息队列中获取信息
                    log.error("处理正常消息队列异常{}", e.getMessage());
                    handlerPendingList();
                }
            }
        }

        /**
         * 处理 正常消息队列中的消息
         * @return
         */
        private Boolean handlerMessageList() {
            //1.获取消息队列中的订单信息 XREADGROUP GROUP g1 c1 COUNT 1 BLOCK 2000 STREAMS stream.orders >
            // string 表示这个消息的id 后面两个是map集合 键值对
            List<MapRecord<String, Object, Object>> message = stringRedisTemplate.opsForStream().read(Consumer.from("g1", "c1"),
                    StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2)),
                    StreamOffset.create("stream.orders", ReadOffset.from(">")));
            // 2.判断订单信息是否为空
            if (message == null || message.isEmpty()) {
                // 如果为null，说明没有消息，继续下一次循环

                return false;
            }
            // 3.解析数据(userId:value , voucherId : value ,id : orderId)
            MapRecord<String, Object, Object> record = message.get(0);
            Map<Object, Object> value = record.getValue();
            VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(value, new VoucherOrder(), true);
            // 4.写入数据库
            handleCreateOrder(voucherOrder);
            // 5.向消息队列发送确认信息
            stringRedisTemplate.opsForStream().acknowledge("stream.orders", "g1", record.getId());
            log.info("处理订单消息成功");
            return true;
        }

        /**
         * 处理 已取但未发送确认的消息队列（即 处理消息的过程出现异常）
         */
        private void handlerPendingList() {
            //1.获取消息队列中（pendingList）的订单信息 XREADGROUP GROUP g1 c1 COUNT 1 BLOCK 2000 STREAMS stream.orders 0
            while (true) {
                try {
                    List<MapRecord<String, Object, Object>> messageList = stringRedisTemplate.opsForStream().read(Consumer.from("g1", "c1")
                            , StreamReadOptions.empty().count(1).block(Duration.ofSeconds(2)),
                            StreamOffset.create("stream.orders", ReadOffset.from("0")));
                    if(messageList ==null || messageList.isEmpty()){
                        break;
                    }
                    // 解析数据
                    MapRecord<String, Object, Object> record = messageList.get(0);
                    Map<Object, Object> value = record.getValue();
                    VoucherOrder voucherOrder = BeanUtil.fillBeanWithMap(value, new VoucherOrder(), true);
                    // 3.创建订单
                    createVoucherOrder(voucherOrder);
                    // 4.确认消息 XACK
                    stringRedisTemplate.opsForStream().acknowledge("stream.orders", "g1", record.getId());
                    log.info("处理未确认订单消息成功");
                } catch (Exception e) {
                   // 处理为未确认消息出错
                    log.error("处理pendingList出错");
                    try {
                        Thread.sleep(20);
                    } catch (InterruptedException ex) {
                        ex.printStackTrace();
                    }
                }
            }
        }

        private void handleCreateOrder(VoucherOrder voucherOrder) {
            Long userId = voucherOrder.getUserId();
            RLock lock = redissonClient.getLock("skill:vocher:" + userId);
            boolean b = lock.tryLock();
            if (!b) {
                // 获取锁失败
                log.error("获取锁失败");
            }
            try {
                // 获取锁成功
                //解决spring事务失效 需添加依赖aspectj 并开启对外暴露代理对象
                proxy.createVoucherOrder(voucherOrder);
                log.info("异步创建订单成功");
            } finally {
                lock.unlock();
            }
        }
    }

    /**
     * 异步方式
     */
    @Override
    @Transactional
    public void createVoucherOrder(VoucherOrder voucherOrder) {
        // 1.一人一单 首先判断用户是否下单
        Long count = voucherOrderMapper.selectCount(new LambdaQueryWrapper<VoucherOrder>()
                .eq(VoucherOrder::getVoucherId, voucherOrder.getVoucherId())
                .eq(VoucherOrder::getUserId, voucherOrder.getUserId()));
        if (count > 0) {
            log.error("该用户已下单");
        }

        SeckillVoucher voucher = seckillVoucherService.getById(voucherOrder.getVoucherId());
        // 2.库存是否充足
        Integer stock = voucher.getStock();
        if (stock < 1) {
            // 2.1不充足直接返回 提示信息
            log.error("库存不足");
        }
        //3.充足库存减一
        boolean flage = seckillVoucherService.update().setSql("stock = stock -1")
                .eq("voucher_id", voucherOrder.getVoucherId())
                //.eq("stock", stock)
                .gt("stock", 0)
                .update();
        if (!flage) {
            log.error("下单失败,库存不足");
            return;
        }
        // 4保存数据库
        voucherOrderMapper.insert(voucherOrder);
    }
//    /**
//     * 创建秒杀券订单  单机
//     * @param voucherId
//     * @return
//     */
//    @Override
//    public Result create(Long voucherId) {
//
//        Long userId = UserHolder.getUser().getId();
//        synchronized (userId.toString().intern()){
//            //解决spring事务失效 需添加依赖aspectfj 并开启对外暴露代理对象
//            IVoucherOrderService o =(IVoucherOrderService) AopContext.currentProxy();
//            return  o.createVoucherOrder(voucherId,userId);
//        }
//    }
    /**
     * 创建秒杀券订单  集群 分布式锁
     * 使用 redis setnx数据结构
     * @param voucherId
     * @return
     */
//    public Result create(Long voucherId) {
//
//        Long userId = UserHolder.getUser().getId();
//        RedisLock redisLock = new RedisLock("skill:vocher" + userId, stringRedisTemplate);
//
//        boolean b = redisLock.tryLock(1000L);
//        if(!b) {
//            // 获取锁失败
//            return Result.fail("不可重复下单");
//        }
//        try {
//            // 获取锁成功
//            //解决spring事务失效 需添加依赖aspectfj 并开启对外暴露代理对象
//            IVoucherOrderService o = (IVoucherOrderService) AopContext.currentProxy();
//            return o.createVoucherOrder(voucherId, userId);
//        } finally {
//            redisLock.unLock();
//        }
//
//    }

    /**
     * 使用 redissession 解决分布式锁
     * @param voucherId
     * @return
     */
//    public Result create(Long voucherId) {
//
//        Long userId = UserHolder.getUser().getId();
//        RLock lock = redissonClient.getLock("skill:vocher" + userId);
//        boolean b = lock.tryLock();
//        if(!b) {
//            // 获取锁失败
//            return Result.fail("不可重复下单");
//        }
//        try {
//            // 获取锁成功
//            //解决spring事务失效 需添加依赖aspectj 并开启对外暴露代理对象
//            IVoucherOrderService o = (IVoucherOrderService) AopContext.currentProxy();
//            return o.createVoucherOrder(voucherId, userId);
//        } finally {
//           lock.unlock();
//        }
//
//    }

    /**
     * 秒杀下单优化
     * 异步写入数据库(基于jvm的阻塞对列)
     */

//    public Result create(Long voucherId) {
//        proxy = (IVoucherOrderService) AopContext.currentProxy();
//        Long userId = UserHolder.getUser().getId();
//        RLock lock = redissonClient.getLock("lock:skill:vocher:" + userId);
//        boolean b = lock.tryLock();
//        if(!b) {
//            // 获取锁失败
//            log.error("获取锁");
//            return Result.fail("不可重复下单");
//        }
//        Long id;
//        try {
//            // 执行lua脚本 判断是否有条件下单
//            Long falge = stringRedisTemplate.execute(CREATE_ORDER, Collections.emptyList(), voucherId.toString(), userId.toString());
//            if (falge != 0) {
//                return Result.fail(falge == 1 ? "库存不足" : "你已经有过一单下单");
//            }
//            id = redisIdWork.nextId(RedisConstants.SKILL_VOUCHER_ORDER_ID);
//            // 封装订单对象
//            VoucherOrder voucherOrder = new VoucherOrder();
//            // 4.1秒杀券ID
//            voucherOrder.setVoucherId(voucherId);
//            // 4.2用户ID
//            voucherOrder.setUserId(userId);
//            // 4.3订单ID
//            voucherOrder.setId(id);
//            // 放入阻塞队列
//            orderQueue.add(voucherOrder);
//        } finally {
//            lock.unlock();
//        }
//        return Result.ok(id);
//    }

    /**
     * 秒杀下单优化
     * 异步写入数据库(基于redis的消费组消息队列 stream)
     */
    @Override
    public Result create(Long voucherId) {
        proxy = (IVoucherOrderService) AopContext.currentProxy();
        Long userId = UserHolder.getUser().getId();
        RLock lock = redissonClient.getLock("lock:skill:vocher:" + userId);
        boolean b = lock.tryLock();
        if (!b) {
            // 获取锁失败
            log.error("获取锁");
            return Result.fail("不可重复下单");
        }
        Long id;
        try {
            // 执行lua脚本 判断是否有条件下单 并将订单信息添加到消息队列中,详细见lua脚本代码
            id = redisIdWork.nextId(RedisConstants.SKILL_VOUCHER_ORDER_ID);
            Long falge = stringRedisTemplate.execute(CREATE_ORDER, Collections.emptyList(), voucherId.toString(), userId.toString(), id.toString());
            if (falge != 0) {
                return Result.fail(falge == 1 ? "库存不足" : "你已经有过一单下单");
            }
        } finally {
            lock.unlock();
        }
        return Result.ok(id);
    }

    /**
     * 同步方式
     *
     * @param voucherId
     * @param userId
     * @return
     */
    @Override
    @Transactional
    public Result createVoucherOrder(Long voucherId, Long userId) {
        // 1.一人一单 首先判断用户是否下单
        Long count = voucherOrderMapper.selectCount(new LambdaQueryWrapper<VoucherOrder>()
                .eq(VoucherOrder::getVoucherId, voucherId)
                .eq(VoucherOrder::getUserId, userId));
        if (count > 0) {
            return Result.fail("该用户已下单");
        }
        /**悲观锁*/
//        skillLock.lock();
//        try {
//            // 1.获取秒杀券信息(对象)
//            SeckillVoucher voucher = seckillVoucherService.getById(voucherId);
//            // 2.库存是否充足
//            Integer stock = voucher.getStock();
//            if(stock<1){
//            // 2.1不充足直接返回 提示信息
//                return Result.fail("库存不足");
//            }
//            // 3.充足库存减一
//            seckillVoucherService.update().setSql("stock = stock -1").eq("voucher_id", voucherId).update();
//        } finally {
//                skillLock.unlock();
//        }

        /** 乐观锁解决超卖*/
        SeckillVoucher voucher = seckillVoucherService.getById(voucherId);
        // 2.库存是否充足
        Integer stock = voucher.getStock();
        if (stock < 1) {
            // 2.1不充足直接返回 提示信息
            return Result.fail("库存不足");
        }
        //3.充足库存减一
        boolean flage = seckillVoucherService.update().setSql("stock = stock -1")
                .eq("voucher_id", voucherId)
                //.eq("stock", stock)
                .gt("stock", 0)
                .update();
        if (!flage) {
            return Result.fail("下单失败,库存不足");
        }

        // 4.创建订单 保存数据库
        VoucherOrder voucherOrder = new VoucherOrder();
        // 4.1秒杀券ID
        voucherOrder.setVoucherId(voucherId);
        // 4.2用户ID
        voucherOrder.setUserId(userId);
        // 4.3订单ID 这里先自己设置  后序使用redis全局生成ID
        Long id = redisIdWork.nextId(RedisConstants.SKILL_VOUCHER_ORDER_ID);
        voucherOrder.setId(id);
        // 4.4保存数据库
        voucherOrderMapper.insert(voucherOrder);
        // 返回订单号
        return Result.ok(id);
    }


}
